﻿using System;
using System.Collections.Generic;
using System.Text;
using MapReduce.Comm;
using MapReduce.Utils;

namespace MapReduce
{
    public class MapReduce
    {

        private ushort mPort;


        public IMap Map = null;
        public IReduce Reduce = null;

        public Callbacks.JobDone SuccessCallback = null;
        public Callbacks.JobError ErrorCallback = null;

        public MapReduce()
        {
            if (SlaveManager.Free.Count == 0)
            {
                throw new Exception("No slaves are available to do the job.");
            }
            mPort = SlaveManager.Port;
        }

        public void Process(IEnumerable<KeyValuePair<String, String>> input, uint pInputCount)
        {
            if (SuccessCallback == null) throw new Exception("MapReduce.SuccessCallback must not be null");
            if (ErrorCallback == null) throw new Exception("MapReduce.ErrorCallback must not be null");

            //mFinalCallback = pCallback;
            Dictionary<String, List<String>> dictMapOutput = new Dictionary<string, List<string>>();


            JobManager<String> mapJob = new JobManager<string>("MapReduce.Services.ExecuteMapService2", Map, mPort);

            mapJob.ErrorCallback = ErrorCallback;


            //Map job from one slave completed. Progress.
            mapJob.ProgressCallback  =delegate(IEnumerable<KeyValuePair<String, String>> pList)
            {

                foreach (KeyValuePair<String, String> tempPair in pList)
                {
                    List<String> vals;
                    if (dictMapOutput.ContainsKey(tempPair.Key))
                    {
                        vals = dictMapOutput[tempPair.Key];
                    }
                    else
                    {
                        vals = new List<String>();
                        dictMapOutput.Add(tempPair.Key, vals);
                    }
 
                    vals.Add(tempPair.Value);
                }
            };
            //End SetProgressCallback



            //Map job done
            mapJob.DoneCallback = delegate(IEnumerable<KeyValuePair<String, String>> pData)
            {
                ProcessReduce(dictMapOutput);
                //"MapReduce.Services.ExecuteReduceService"
            };

            mapJob.Process(input, pInputCount);
        }


        private void ProcessReduce(Dictionary<String, List<String>> pDictMapOutput)
        {
            List<KeyValuePair<string, string>> listReduceOutput = new List<KeyValuePair<string, string>>();

            JobManager<List<String>> job = new JobManager<List<String>>("MapReduce.Services.ExecuteReduceService2", Reduce, mPort);

            job.ErrorCallback = ErrorCallback;

            //Reduce job from one slave completed. Progress.
            job.ProgressCallback = delegate(IEnumerable<KeyValuePair<String, String>> pList)
            {

                foreach (KeyValuePair<String, String> outPair in pList)
                {
                    listReduceOutput.Add(outPair);
                }
            };


            //Reduce job done
            job.DoneCallback = delegate(IEnumerable<KeyValuePair<String, String>> pData)
            {
                SuccessCallback(listReduceOutput);
            };

            job.Process(pDictMapOutput, (uint)pDictMapOutput.Count);
        }



    }//MapReduce
}
